package cn.doitedu.feature_etl;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/9
 * @Desc: 学大数据，上多易教育
 *  把 kafka中的  广告引擎后端日志：广告请求特征日志 ，  同步到hbase
 **/
public class Job02_AdRequestLog_Hbase {


    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1 ms");

        // 1. 建表，映射kafka中的 广告请求特征日志
        tenv.executeSql("create table request_log_kfk(\n" +
                " line string\n" +
                ") WITH (                     \n" +
                "    'connector' = 'kafka',   \n" +
                "    'topic' = 'ad-request-log',  \n" +
                "    'properties.bootstrap.servers' = 'doitedu:9092', \n" +
                "    'properties.group.id' = 'doit44_g2',             \n" +
                "    'scan.startup.mode' = 'latest-offset',           \n" +
                "    'value.format' = 'csv',                         \n" +
                "    'value.csv.field-delimiter' = U&'\\0001'        \n" +
                ") ");


       // tenv.executeSql("select line from request_log_kfk").print();




        // 2. 建表，映射hbase中的 广告请求特征日志表
        tenv.executeSql(
                " create table sink_hbase (                      "+
                        "   ad_tracking_id string                 "+
                        "   ,f ROW<features string>               "+
                        " ) WITH (                                "+
                        "  'connector' = 'hbase-2.2',             "+
                        "  'table-name' = 'ad_request_log',         "+
                        "  'zookeeper.quorum' = 'doitedu:2181'    "+
                        " )                                        "
        );



        // 3. insert ... select ...
        tenv.executeSql("insert into sink_hbase " +
                "select " +
                "  JSON_VALUE(line,'$.ad_tracking_id') as ad_tracking_id ," +
                "  row(line) as f " +
                "from request_log_kfk ");



    }


}
